package com.wuwangfu.func;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-06-28
 */
public class QueryStateRichMapFunc extends RichMapFunction<Tuple2<String,Integer>,Tuple2<String,Integer>> {

    //使用ValueState保存单词对应的次数
    private transient ValueState<Integer> countState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化数据状态或恢复历史状态数据
        ValueStateDescriptor<Integer> stateDesctor = new ValueStateDescriptor<>(
                "word-state",//指定状态描述器的名称
                Integer.class);//存储数据的类型
        //设置状态可以查询，并指定状态查询名称
        stateDesctor.setQueryable("query-word-count");
        //获取 getRuntimeContext 并根据状态描述器取出对应的state
        countState = getRuntimeContext().getState(stateDesctor);
    }

    @Override
    public Tuple2<String, Integer> map(Tuple2<String, Integer> tuple) throws Exception {
        //获取当前输入单词次数
        Integer current = tuple.f1;
        //获取历史次数
        Integer counts = countState.value();
        //历史次数是否为空
        if (counts == null){
            //如果历史次数为空，初始值设置为0
            counts = 0;
        }
        //将当前的次数和历史此事件进行累加
        int total = current + counts;
        //更新状态
        countState.update(total);
        //将累加后的次数放入到tuple中，进行返回
        tuple.f1 = total;
        return tuple;
    }
}
